package com.gin.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.{breakOut, mutable}
import scala.util.control.Breaks

object L07_RddOver {

  def main(args: Array[String]): Unit = {
    //加载配置 获取spark上下文对象
    val conf = new SparkConf().setMaster("local").setAppName("L01")
    val sc = new SparkContext(conf)
    //只打印错误日志
    sc.setLogLevel("ERROR")

    val fileRDD: RDD[String] = sc.textFile("scala/data/tqdata")

    //topN 分组取TopN(二次排序)
    //2019-6-1	39
    //2019-6-1	38
    //同月份中 温度最高的两天

    val data = fileRDD.map(line => line.split("\t")).map(
      arr => {
        val dateArr = arr(0).split("-")
        //(年,月,日, 温度)
        (dateArr(0).toInt, dateArr(1).toInt, dateArr(2).toInt, arr(1).toInt)
      }
    )

    //方式一:
    //先分组
    //用了groupByKey   mutable.HashMap在当月数据量很大情况下容易OOM
    val groupData = data.map(tuple4 => ((tuple4._1, tuple4._2), (tuple4._3, tuple4._4)))
      .groupByKey()

    //年月: [(日1,温度1), (日2,温度2)]
    val res = groupData.mapValues(
      //获取values进行处理
      arr => {
        //(日1,温度1)
        val map = new mutable.HashMap[Int, Int]()
        //map中存储最大温度(日, 当日最大温度), 对每日温度去重取最大值
        arr.foreach(x => {
          if (map.get(x._1).getOrElse(0) < x._2) map.put(x._1, x._2)
        })

        map.toList.sorted(new Ordering[(Int, Int)] {
          //倒序排序
          override def compare(x: (Int, Int), y: (Int, Int)): Int = y._2.compareTo(x._2)
        })
      }
    )

    res.foreach(println)


    println()
    println("-------- partition test 4 --------")
    println()


    // 方式二, 减少内存的使用
    //分布式计算的核心思想：调优天下无敌：combineByKey
    //分布式是并行的，离线批量计算有个特征就是后续步骤(stage)依赖其一步骤(stage)
    //如果前一步骤(stage)能够加上正确的combineByKey
    //我们自定的combineByKey的函数，是尽量压缩内存中的数据
    val newKv = data.map(tuple4 => ((tuple4._1, tuple4._2), (tuple4._3, tuple4._4)))

    // 使用 combineByKey 自定义分组过程
    val res2: RDD[((Int, Int), Array[(Int, Int)])] = newKv.combineByKey(
      //createCombiner: V => C,
      //第一条记录的 value,在key对应的hashmap对象中如何存储
      //仅放3个元素的数组, idx=0: 最大温度的(日, 当日最大温度)二元组
      (value: (Int, Int)) => { Array(value, (0,0), (0,0)) },
      //mergeValue: (C, V) => C,
      //如果有第二条记录,第二条以及以后的value,如果放入到历史的hashmap对象中
      (oldVal: Array[(Int, Int)], newVal: (Int, Int)) => {
        //去重，排序
        var flg = -1
        //新进来的元素特征：
        //  日相同  1）温度大 2）温度小
        //  日不同  1）温度大 2）温度小
        val loop = new Breaks
        loop.breakable {
          for (i <- 0 until oldVal.length) {
            if (oldVal(i)._1 == newVal._1) {
              if (oldVal(i)._2 < newVal._2) {
                //新的温度大, 更新
                oldVal(i) = newVal
              }
              loop.break
            } else {
              if (oldVal(i)._2 < newVal._2) {
                //新的温度大, 数组移动
                flg = i
                loop.break
              }
            }
          }
        }

        //更新温度排序
        if (flg >= 0 && flg <= 2) {
          val range = Range(oldVal.length-1, flg-1, -1)
          for (i <- range) {
            if (i == flg) {
              //替换为新温度
              oldVal(i) = newVal
            } else {
              //比新温度小的后移一位
              oldVal(i) = oldVal(i - 1)
            }
          }
        }

        println("oldVal=", oldVal(0), oldVal(1), oldVal(2))
        oldVal
      },
      //mergeCombiners: (C, C) => C,
      //合并溢写结果的函数: 每个分区的二元组元素如何进行合并(reduce)
      (v1: Array[(Int, Int)], v2: Array[(Int, Int)]) => {
        //关注去重
        val union: Array[(Int, Int)] = v1.union(v2)
        union.sorted
      }
    )

    res2.map(x=>(x._1,x._2.toList)).foreach(println)



  }

}
